package com.donglink.test.logmonitor;

import com.donglink.test.logmonitor.bolt.EstimateBolt;
import com.donglink.test.logmonitor.bolt.FilterBolt;
import com.donglink.test.logmonitor.bolt.MessageBolt;
import com.donglink.test.logmonitor.dao.MybatisDriver;
import com.donglink.test.logmonitor.spout.InputSpout;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;

public class LogMonitorDriver {
    public static void main(String[] args) {
        MybatisDriver.init();
        TopologyBuilder builder=new TopologyBuilder();
        //输入系统信息spout
        builder.setSpout("kafkaSpout",new InputSpout(),2);
        //过滤掉不是规则中的信息
        builder.setBolt("filterBolt",new FilterBolt(),2).globalGrouping("kafkaSpout");
        //判断是否符合规则
        builder.setBolt("estimateBolt",new EstimateBolt(),2).globalGrouping("filterBolt");
        //将符合规则的信息发送到邮件
        builder.setBolt("messageBolt",new MessageBolt(),2).globalGrouping("estimateBolt");



        //创建一个config用于创建toplogy
        Config config =new Config();
        config.setNumWorkers(3);

        //集群
        // StormSubmitter.submitTopology("myFirst",config,builder.createTopology());


        //本地模式
        LocalCluster localCluster=new LocalCluster();

        localCluster.submitTopology("12",config,builder.createTopology());
    }
}
